package com.enhanced.mqtt.client;

import org.eclipse.paho.mqttv5.client.*;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
import org.eclipse.paho.mqttv5.common.packet.UserProperty;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Scanner;
import java.util.UUID;

/**
 * MQTT 5.0客户端示例
 * 演示MQTT 5.0的特性，如用户属性、请求/响应、主题别名等
 * 注意：需要添加Paho MQTT v5客户端依赖
 */
public class Mqtt5Client {
    
    private static final String BROKER_URL = "tcp://192.168.3.234:1883";
    private static final String TOPIC = "test/mqtt5";
    private static final String RESPONSE_TOPIC = "test/response";
    private static final int QOS = 1;
    
    public static void main(String[] args) {
        try {
            runDemo(new Scanner(System.in));
        } catch (Exception e) {
            System.out.println("MQTT错误: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 运行演示，使用外部传入的Scanner
     * @param scanner 已经打开的Scanner
     */
    public static void runDemo(Scanner scanner) throws Exception {
        System.out.println("=== MQTT 5.0客户端演示 ===");
        System.out.println("注意：此演示需要EMQX支持MQTT 5.0协议");
        
        System.out.println("请选择要演示的MQTT 5.0特性：");
        System.out.println("1. 用户属性");
        System.out.println("2. 请求/响应模式");
        System.out.println("3. 消息过期间隔");
        System.out.println("4. 主题别名");
        System.out.print("请选择(1-4): ");
        
        int choice = scanner.nextInt();
        scanner.nextLine(); // 消费换行符
        
        switch (choice) {
            case 1:
                demonstrateUserProperties();
                break;
            case 2:
                demonstrateRequestResponse();
                break;
            case 3:
                demonstrateMessageExpiry();
                break;
            case 4:
                demonstrateTopicAlias();
                break;
            default:
                System.out.println("无效选择！");
        }
        
        // 不关闭scanner，由调用者负责关闭
    }
    
    /**
     * 演示用户属性
     * 用户属性允许在消息中添加键值对元数据
     */
    private static void demonstrateUserProperties() throws MqttException, InterruptedException {
        System.out.println("=== 用户属性演示 ===");
        
        // 创建客户端
        String clientId = "MQTT5Client-" + UUID.randomUUID().toString().substring(0, 8);
        MqttAsyncClient client = new MqttAsyncClient(BROKER_URL, clientId, new MemoryPersistence());
        
        // 设置回调
        client.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse disconnectResponse) {
                System.out.println("已断开连接: " + disconnectResponse.getReasonString());
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.out.println("MQTT错误: " + exception.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("\n收到消息:");
                System.out.println("  主题: " + topic);
                System.out.println("  内容: " + new String(message.getPayload(), StandardCharsets.UTF_8));
                System.out.println("  QoS: " + message.getQos());
                
                // 打印用户属性
                if (message.getProperties() != null && message.getProperties().getUserProperties() != null) {
                    System.out.println("  用户属性:");
                    for (UserProperty prop : message.getProperties().getUserProperties()) {
                        System.out.println("    " + prop.getKey() + ": " + prop.getValue());
                    }
                }
            }

            @Override
            public void deliveryComplete(IMqttToken token) {
                System.out.println("消息发送完成");
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println(reconnect ? "已重新连接到服务器" : "已连接到服务器");
            }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) {
                System.out.println("收到认证包: " + reasonCode);
            }
        });
        
        // 连接服务器
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setCleanStart(true);
        System.out.println("正在连接到MQTT 5.0服务器: " + BROKER_URL);
        client.connect(options).waitForCompletion(5000);
        
        // 订阅主题
        client.subscribe(TOPIC, QOS).waitForCompletion();
        System.out.println("已订阅主题: " + TOPIC);
        
        // 创建消息并添加用户属性
        String payload = "这是一条带有用户属性的MQTT 5.0消息";
        MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
        message.setQos(QOS);
        
        MqttProperties properties = new MqttProperties();
        properties.setUserProperties(Arrays.asList(new UserProperty[]{
                new UserProperty("创建时间", String.valueOf(System.currentTimeMillis())),
                new UserProperty("来源", "Java客户端"),
                new UserProperty("应用", "MQTT 5.0演示")
        }));
        message.setProperties(properties);
        
        // 发布消息
        client.publish(TOPIC, message).waitForCompletion();
        System.out.println("已发布带用户属性的消息");
        
        // 等待接收消息
        System.out.println("等待5秒接收消息...");
        Thread.sleep(5000);

        // 断开连接
        client.disconnect().waitForCompletion();
        System.out.println("已断开连接");
    }
    
    /**
     * 演示请求/响应模式
     * 通过响应主题和相关数据实现同步请求/响应模式
     */
    private static void demonstrateRequestResponse() throws MqttException, InterruptedException {
        System.out.println("=== 请求/响应模式演示 ===");
        
        // 创建客户端A（请求者）
        String clientIdA = "MQTT5RequestClient";
        MqttAsyncClient clientA = new MqttAsyncClient(BROKER_URL, clientIdA, new MemoryPersistence());
        
        // 创建客户端B（响应者）
        String clientIdB = "MQTT5ResponseClient";
        MqttAsyncClient clientB = new MqttAsyncClient(BROKER_URL, clientIdB, new MemoryPersistence());
        
        // 设置客户端B的回调（响应请求）
        clientB.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse disconnectResponse) {
                System.out.println("客户端B已断开连接");
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.out.println("客户端B错误: " + exception.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("客户端B收到请求: " + new String(message.getPayload(), StandardCharsets.UTF_8));
                
                // 获取响应主题和相关数据
                String responseTopic = message.getProperties().getResponseTopic();
                byte[] correlationData = message.getProperties().getCorrelationData();
                
                if (responseTopic != null) {
                    // 创建响应消息
                    String responsePayload = "这是对请求的响应";
                    MqttMessage responseMessage = new MqttMessage(responsePayload.getBytes(StandardCharsets.UTF_8));
                    responseMessage.setQos(QOS);
                    
                    // 设置相同的相关数据以关联请求和响应
                    MqttProperties properties = new MqttProperties();
                    properties.setCorrelationData(correlationData);
                    responseMessage.setProperties(properties);
                    
                    // 发送响应
                    clientB.publish(responseTopic, responseMessage).waitForCompletion();
                    System.out.println("客户端B已发送响应");
                }
            }

            @Override
            public void deliveryComplete(IMqttToken token) {
                System.out.println("客户端B消息发送完成");
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("客户端B" + (reconnect ? "已重新连接" : "已连接"));
            }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) {
                // 不处理
            }
        });
        
        // 设置客户端A的回调（接收响应）
        clientA.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse disconnectResponse) {
                System.out.println("客户端A已断开连接");
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.out.println("客户端A错误: " + exception.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("客户端A收到响应: " + new String(message.getPayload(), StandardCharsets.UTF_8));
                
                // 检查相关数据
                if (message.getProperties().getCorrelationData() != null) {
                    String correlationId = new String(message.getProperties().getCorrelationData(), StandardCharsets.UTF_8);
                    System.out.println("相关数据ID: " + correlationId);
                }
            }

            @Override
            public void deliveryComplete(IMqttToken token) {
                System.out.println("客户端A消息发送完成");
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("客户端A" + (reconnect ? "已重新连接" : "已连接"));
            }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) {
                // 不处理
            }
        });
        
        // 连接客户端B（响应者）
        MqttConnectionOptions optionsB = new MqttConnectionOptions();
        optionsB.setCleanStart(true);
        clientB.connect(optionsB).waitForCompletion();
        clientB.subscribe(TOPIC, QOS).waitForCompletion();
        System.out.println("响应者客户端已连接并订阅主题: " + TOPIC);
        
        // 连接客户端A（请求者）
        MqttConnectionOptions optionsA = new MqttConnectionOptions();
        optionsA.setCleanStart(true);
        clientA.connect(optionsA).waitForCompletion();
        clientA.subscribe(RESPONSE_TOPIC, QOS).waitForCompletion();
        System.out.println("请求者客户端已连接并订阅响应主题: " + RESPONSE_TOPIC);
        
        // 创建请求消息
        String requestPayload = "这是一个请求消息";
        MqttMessage requestMessage = new MqttMessage(requestPayload.getBytes(StandardCharsets.UTF_8));
        requestMessage.setQos(QOS);
        
        // 设置响应主题和相关数据
        MqttProperties properties = new MqttProperties();
        properties.setResponseTopic(RESPONSE_TOPIC);
        String correlationId = "req-" + System.currentTimeMillis();
        properties.setCorrelationData(correlationId.getBytes(StandardCharsets.UTF_8));
        requestMessage.setProperties(properties);
        
        // 发送请求
        clientA.publish(TOPIC, requestMessage).waitForCompletion();
        System.out.println("请求者已发送请求，相关ID: " + correlationId);
        
        // 等待响应
        System.out.println("等待响应...");
        Thread.sleep(5000);
        
        // 断开连接
        clientA.disconnect().waitForCompletion();
        clientB.disconnect().waitForCompletion();
        System.out.println("请求/响应演示结束");
    }
    
    /**
     * 演示消息过期间隔
     * 允许设置消息在服务器上存在的最大时间
     */
    private static void demonstrateMessageExpiry() throws MqttException, InterruptedException {
        System.out.println("=== 消息过期间隔演示 ===");
        
        // 创建发布者客户端
        String publisherClientId = "MQTT5ExpiryPublisher";
        MqttAsyncClient publisher = new MqttAsyncClient(BROKER_URL, publisherClientId, new MemoryPersistence());
        
        // 连接服务器
        MqttConnectionOptions publisherOptions = new MqttConnectionOptions();
        publisherOptions.setCleanStart(true);
        publisher.connect(publisherOptions).waitForCompletion();
        System.out.println("发布者已连接");
        
        // 发布三条不同过期间隔的消息
        MqttMessage msg1 = new MqttMessage("这条消息将在5秒后过期".getBytes(StandardCharsets.UTF_8));
        msg1.setQos(QOS);
        MqttProperties props1 = new MqttProperties();
        props1.setMessageExpiryInterval(5L); // 5秒过期
        msg1.setProperties(props1);
        publisher.publish(TOPIC, msg1).waitForCompletion();
        System.out.println("已发布5秒过期消息");
        
        MqttMessage msg2 = new MqttMessage("这条消息将在15秒后过期".getBytes(StandardCharsets.UTF_8));
        msg2.setQos(QOS);
        MqttProperties props2 = new MqttProperties();
        props2.setMessageExpiryInterval(15L); // 15秒过期
        msg2.setProperties(props2);
        publisher.publish(TOPIC, msg2).waitForCompletion();
        System.out.println("已发布15秒过期消息");
        
        MqttMessage msg3 = new MqttMessage("这条消息没有过期时间".getBytes(StandardCharsets.UTF_8));
        msg3.setQos(QOS);
        publisher.publish(TOPIC, msg3).waitForCompletion();
        System.out.println("已发布无过期时间消息");
        
        // 断开发布者
        publisher.disconnect().waitForCompletion();
        System.out.println("发布者已断开连接");
        
        // 等待一段时间让第一条消息过期
        System.out.println("等待10秒后连接订阅者...");
        Thread.sleep(10000);
        
        // 创建订阅者客户端
        String subscriberClientId = "MQTT5ExpirySubscriber";
        MqttAsyncClient subscriber = new MqttAsyncClient(BROKER_URL, subscriberClientId, new MemoryPersistence());
        
        // 设置订阅者回调
        subscriber.setCallback(new MqttCallback() {
            @Override
            public void disconnected(MqttDisconnectResponse disconnectResponse) {
                System.out.println("订阅者已断开连接");
            }

            @Override
            public void mqttErrorOccurred(MqttException exception) {
                System.out.println("订阅者错误: " + exception.getMessage());
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                System.out.println("收到消息: " + new String(message.getPayload(), StandardCharsets.UTF_8));
                
                // 打印剩余过期时间
                if (message.getProperties() != null && message.getProperties().getMessageExpiryInterval() != null) {
                    System.out.println("  剩余过期时间: " + message.getProperties().getMessageExpiryInterval() + "秒");
                } else {
                    System.out.println("  没有过期时间");
                }
            }

            @Override
            public void deliveryComplete(IMqttToken token) {
                // 不处理
            }

            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                System.out.println("订阅者已连接");
            }

            @Override
            public void authPacketArrived(int reasonCode, MqttProperties properties) {
                // 不处理
            }
        });
        
        // 连接订阅者
        MqttConnectionOptions subscriberOptions = new MqttConnectionOptions();
        subscriberOptions.setCleanStart(true);
        subscriber.connect(subscriberOptions).waitForCompletion();
        
        // 订阅主题
        subscriber.subscribe(TOPIC, QOS).waitForCompletion();
        System.out.println("订阅者已订阅主题: " + TOPIC);
        
        // 等待接收消息
        System.out.println("等待10秒接收未过期的消息...");
        Thread.sleep(10000);
        
        // 断开订阅者
        subscriber.disconnect().waitForCompletion();
        System.out.println("订阅者已断开连接");
        System.out.println("消息过期间隔演示结束");
    }
    
    /**
     * 演示主题别名
     * 主题别名可以减少网络传输中的主题长度
     */
    private static void demonstrateTopicAlias() throws MqttException, InterruptedException {
        System.out.println("=== 主题别名演示 ===");
        
        // 创建客户端
        String clientId = "MQTT5TopicAliasClient";
        MqttAsyncClient client = new MqttAsyncClient(BROKER_URL, clientId, new MemoryPersistence());
        
        // 设置连接选项，启用主题别名
        MqttConnectionOptions options = new MqttConnectionOptions();
        options.setCleanStart(true);
        options.setTopicAliasMaximum(10); // 设置主题别名的最大数量
        
        // 连接服务器
        System.out.println("正在连接到服务器并启用主题别名功能");
        client.connect(options).waitForCompletion();
        System.out.println("已连接到服务器");
        
        // 订阅主题
        client.subscribe(TOPIC, QOS).waitForCompletion();
        System.out.println("已订阅主题: " + TOPIC);
        
        // 发布多条消息到相同的主题
        System.out.println("发布5条消息到相同主题，应使用主题别名减少后续消息大小");
        
        for (int i = 1; i <= 5; i++) {
            MqttMessage message = new MqttMessage(("主题别名测试消息 #" + i).getBytes(StandardCharsets.UTF_8));
            message.setQos(QOS);
            client.publish(TOPIC, message).waitForCompletion();
            System.out.println("已发布消息 #" + i);
            Thread.sleep(1000);
        }
        
        // 断开连接
        client.disconnect().waitForCompletion();
        System.out.println("已断开连接");
        System.out.println("主题别名演示结束");
        System.out.println("注意：主题别名是在协议层面实现的，可以通过网络抓包工具观察到实际效果");
    }
} 